package com.galeno.day08;

import lombok.AllArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.connector.jdbc.JdbcSink;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2517:47
 */
public class JDBCSink {
//    @AllArgsConstructor
//    static class Student {
//        final String name;
//        final int age;
//    }
    public static void main(String[] args) throws Exception {


        Configuration conf = new Configuration();
        conf.setInteger("rest.port",22222);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        DataStreamSource<String> lines = env.socketTextStream("192.168.77.3", 9999);
        SingleOutputStreamOperator<Student> map = lines.map(new MapFunction<String, Student>() {
            @Override
            public Student map(String value) throws Exception {
                String[] people = value.split(",");

                return new Student(people[0],Integer.parseInt(people[1]));
            }
        });
        map.addSink(JdbcSink.sink("insert into student(name,age)values(?,?)",
                ((preparedStatement, student) -> {
                    preparedStatement.setString(1,student.getName());
                    preparedStatement.setInt(2,student.getAge());
                }),
                JdbcExecutionOptions.builder()
                        .withBatchSize(1)
                        .withMaxRetries(2)
                        .withMaxRetries(10000)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://localhost:3306/oa")
                .withDriverName("com.mysql.cj.jdbc.Driver")
                .withUsername("root")
                .withPassword("root")
                .build()
                ));
        env.execute();




    }
}
